package com.king.app.ods

import com.king.app.function.CustomerDeseriallization
import com.king.util.MyKafkaUtil
import com.ververica.cdc.connectors.mysql.MySqlSource
import com.ververica.cdc.connectors.mysql.table.StartupOptions
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment

/**
 * @Author: KingWang
 * @Date: 2021/12/26  
 * @Desc:
 **/
object FlinkCDC {

  def main(args: Array[String]): Unit = {

    //1. 获取执行环境
    val env = StreamExecutionEnvironment.getExecutionEnvironment
    env.setParallelism(1)

    //1.1 开启ck并指定状态后端fs
//    env.setStateBackend(new FsStateBackend("hdfs://hadoop200:8020/gmall-flink-210325/ck"))
//      .enableCheckpointing(10000L) //头尾间隔：每10秒触发一次ck
//      env.getCheckpointConfig.setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE)  //
//      env.getCheckpointConfig.setCheckpointTimeout(10000L)
//    env.getCheckpointConfig.setMaxConcurrentCheckpoints(2)
//    env.getCheckpointConfig.setMinPauseBetweenCheckpoints(3000l)  //尾和头间隔时间3秒

//    env.setRestartStrategy(RestartStrategies.fixedDelayRestart(3, 10000L));



    //2. 通过flinkCDC构建SourceFunction并读取数据
    val sourceFunction = MySqlSource.builder[String]()
        .hostname("hadoop200")
        .port(3306)
        .username("root")
        .password("root")
        .databaseList("gmall-210325-flink")

        //如果不添加该参数，则消费指定数据库中所有表的数据
        //如果添加，则需要按照 数据库名.表名 的格式指定，多个表使用逗号隔开
//        .tableList("gmall-210325-flink.base_trademark")
        .deserializer(new CustomerDeseriallization())

        //监控的方式：
      // 1. initial 初始化全表拷贝，然后再比较
      // 2. earliest 不做初始化，只从当前的
      // 3. latest  指定最新的
      // 4. specificOffset 指定offset
      // 3. timestamp 比指定的时间大的

        .startupOptions(StartupOptions.latest())
        .build()

    val  dataStream = env.addSource(sourceFunction)

    //3. sink, 写入kafka
    dataStream.print()
    val sinkTopic = "ods_base_db"
    dataStream.addSink(MyKafkaUtil.getKafkaProducer(sinkTopic))
    //4. 启动任务
    env.execute("flinkCDC")

  }

}
